package com.sys.tdhclient.utils

import java.io.File
import java.util.Properties

import com.sys.tdhclient.client.InitClient
import com.sys.tdhclient.startapp.SparkRdHyperbase.properties
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.hbase.{HBaseConfiguration, HTableDescriptor, TableName}
import org.apache.hadoop.hbase.client.{Connection, ConnectionFactory, HBaseAdmin, HTable}
import org.apache.hadoop.security.UserGroupInformation

object HBaseUtils {
  /**
    * 设置HBaseConfiguration
    * @param quorum
    * @param port
    * @param tableName
    */
  private val properties: Properties = SparkSc.getProperties()
  private val quorum: String = properties.getProperty("quorum")
  private val port: String = properties.getProperty("port")
  private val principal: String = properties.getProperty("userPrincipal")
  private val znodeParent: String = properties.getProperty("znodeParent")
  private val confFilePath = System.getProperty("user.dir") + File.separator + "conf" + File.separator
  private val kerberosConfPath = confFilePath+"krb5.conf"
  private val keytabPath = confFilePath+"keytab"
  def getHBaseConfiguration() = {
    val conf = HBaseConfiguration.create
    conf.set("hbase.zookeeper.quorum", quorum)
    conf.set("hbase.zookeeper.property.clientPort", port)
    conf.set("zookeeper.znode.parent", znodeParent)
    conf.setInt("hbase.rpc.timeout", 20000)
    conf.setInt("hbase.client.operation.timeout", 30000)
    conf.setInt("hbase.client.scanner.timeout.period", 200000)
    conf.set("hadoop.security.authentication", "kerberos")
    conf.set("hbase.security.authentication", "kerberos")
    System.setProperty("javax.security.auth.useSubjectCredsOnly", "false")
    System.setProperty("java.security.krb5.conf", kerberosConfPath)
    conf
  }
  @throws[Exception]
  def getConnection: Connection = {
    var connection: Connection = null
    val conf = getHBaseConfiguration
    UserGroupInformation.setConfiguration(conf)
    UserGroupInformation.loginUserFromKeytab(principal, keytabPath)
    connection = ConnectionFactory.createConnection(conf)
    connection
  }

  /**
    * 返回或新建HBaseAdmin
    * @param conf
    * @param tableName
    * @return
    */
  def getHBaseAdmin(conf:Configuration,tableName:String) = {
    val admin = new HBaseAdmin(conf)
    if (!admin.isTableAvailable(tableName)) {
      val tableDesc = new HTableDescriptor(TableName.valueOf(tableName))
      admin.createTable(tableDesc)
    }
    admin
  }

  /**
    * 返回HTable
    * @param conf
    * @param tableName
    * @return
    */
  def getTable(conf:Configuration,tableName:String) = {
    new HTable(conf,tableName)
  }




}
